//import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
//import com.alibaba.ververica.cdc.connectors.mysql.table.StartupOptions;
//import com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction;
//import com.alibaba.ververica.cdc.debezium.StringDebeziumDeserializationSchema;
//import org.apache.flink.api.common.restartstrategy.RestartStrategies;
//import org.apache.flink.runtime.state.filesystem.FsStateBackend;
//import org.apache.flink.streaming.api.CheckpointingMode;
//import org.apache.flink.streaming.api.datastream.DataStreamSource;
//import org.apache.flink.streaming.api.environment.CheckpointConfig;
//import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
//
//public class FlinkCDC {
//    public static void main(String[] args) {
//        //1.创建执行环境
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//
//        //2.Flink-CDC将读取binlog的位置信息以状态的方式保存在CK,如果想要做到断点续传,需要从Checkpoint或者Savepoint启动程序
//        //2.1 开启Checkpoint,每隔5秒钟做一次CK
//        env.enableCheckpointing(5000L);
////        设置超时时间
//        env.getCheckpointConfig().setCheckpointTimeout(10000L);
////        设置最大当前ck数量
//        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
////        设置最小ck间隔
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000);
//        //2.2 指定CK的一致性语义
//        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
//        //2.3 设置任务关闭的时候保留最后一次CK数据
//        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
//        //2.4 指定从CK自动重启策略
//        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 2000L));
//        //2.5 设置状态后端
//        env.setStateBackend(new FsStateBackend("hdfs://djcluster/flinkCDC"));
//        //2.6 设置访问HDFS的用户名
//        System.setProperty("HADOOP_USER_NAME", "atguigu");
//
//        //3.创建Flink-MySQL-CDC的Source
//        DebeziumSourceFunction<String> mysqlSource = MySQLSource.<String>builder()
//                .hostname("hadoop101")
//                .port(3306)
//                .username("root")
//                .password("000000")
//                .databaseList("gmall2021")
//                .tableList("gmall2021.activity_sku")
//                .deserializer(new StringDebeziumDeserializationSchema())
//                .startupOptions(StartupOptions.initial())
//                .build();
//
//        //4.使用CDC Source从MySQL读取数据
//        DataStreamSource<String> mysqlDS = env.addSource(mysqlSource);
//
//        //5.打印数据
//        mysqlDS.print();
//
//        //6.执行任务
//        try {
//            env.execute();
//        } catch (Exception e) {
//            e.printStackTrace();
//        }
//    }
//}
